package cn.hfuu.spark;

import org.apache.spark.sql.*;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import org.apache.spark.sql.types.*;

/**
 * @author oldliu
 * @since 1.0
 */
public class CountCustomersSchema {
    public static void main(String[] args) throws AnalysisException {
        // 创建SparkSession
        SparkSession spark = SparkSession.builder()
                .appName("SparkApp")
                .master("local")
                .getOrCreate();
        // 假设有一个结构类型
        StructType schema = new StructType(new StructField[]{
                // 定义字段
                DataTypes.createStructField("customer_id", DataTypes.LongType, true),
                DataTypes.createStructField("customer_fname", DataTypes.StringType, true),
                DataTypes.createStructField("customer_lname", DataTypes.StringType, true),
                DataTypes.createStructField("customer_password", DataTypes.StringType, true),
                DataTypes.createStructField("customer_street", DataTypes.StringType, true),
                DataTypes.createStructField("customer_city", DataTypes.StringType, true),
                DataTypes.createStructField("customer_state", DataTypes.StringType, true),
                DataTypes.createStructField("customer_zipcode", DataTypes.StringType, true)
        });


        // 创建JavaSparkContext
        //JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
        Dataset<Row> csv = spark.read()
                .option("header", "false") // 假设CSV没有标题行
                .schema(schema)
                .csv("file:///usr/local/customers.csv");// file:// 本地文件
        csv.createTempView("customers");
        Dataset<Row> sql = csv.sqlContext().sql(" with t as (select count(customer_id)as c,customer_state from " +
                "customers \n" +
                "group by customer_state )\n" +
                "select * from customers  where customer_state in(\n" +
                "select customer_state from t where t.c=(\n" +
                "  select max(c) from t\n" +
                ") )");
        //统计各个州有多少客户
        sql.limit(20).show();
        System.out.println("---------------------");
        sql.foreach(x -> {
            System.out.println(x.getLong(0) + "--" + x.getString(1) + " " + x.getString(2) + "---" + x.getString(6));
        });
        spark.stop();
    }
}
